package com.sui.bigdata.flink.sql.core.util;

import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;

import java.io.File;
import java.util.Enumeration;
import java.util.List;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;

/**
 * @author YongChen
 * @date 2019/12/4 15:44
 * @description
 * @email yong_chen@sui.com
 */
public class ScanUdfUtils {

    final static String UDF_PACKAGE = "com/sui/bigdata/flink/sql/function";
    final static String SCALAR = "scalar";
    final static String TABLE = "table";
    final static String AGGREGATE = "aggregate";
    public static String getUdfs(String udfDirPath) throws Exception {
        StringBuilder udfs = new StringBuilder();
        udfs.append(";");
        File files = new File(udfDirPath);
        for (File file : files.listFiles()) {
            if (file.getAbsolutePath().endsWith("jar")){
            JarFile jarFile = new JarFile(file.getAbsolutePath());
            Enumeration<JarEntry> entrys = jarFile.entries();
            while (entrys.hasMoreElements()) {
                JarEntry jarEntry = entrys.nextElement();
                String className = jarEntry.getName();
                if (className.contains(UDF_PACKAGE) && className.contains("class")) {
                    className = className.replace(".class", "");
                    String funName = className.split("/")[className.split("/").length - 1];
                    className = className.replace("/", ".");
                    if (className.contains(SCALAR)) {
                        udfs.append("create " + SCALAR + " function " + funName + " with " + className + ";");
                    }
                    if (className.contains(TABLE)) {
                        udfs.append("create " + TABLE + " function " + funName + " with " + className + ";");
                    }
                    if (className.contains(AGGREGATE)) {
                        udfs.append("create " + AGGREGATE + " function " + funName + " with " + className + ";");
                    }
                }
            }

        }
        }
       return udfs.toString();
    }

}
